package com.spi.config;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Set;

//@Component
//@EnableScheduling
public class DLQMonitor {

    @Value("${rocketmq.name-server}")
    private String nameServerAddr;

    private static final String DLQ_TOPIC = "%DLQ%consumer-group3"; // 替换为你的 DLQ 主题

    private DefaultMQPullConsumer consumer;

    @PostConstruct
    public void init() throws Exception {
        consumer = new DefaultMQPullConsumer("DLQMonitorGroup");
        consumer.setNamesrvAddr(nameServerAddr);
        consumer.start();
    }

    //@Scheduled(fixedRate = 5000) // 每分钟检查一次
    public void monitorDLQ() {
        try {
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(DLQ_TOPIC);

            for (MessageQueue mq : mqs) {
                long offset = consumer.fetchConsumeOffset(mq, false);
                if (offset < 0) {
                    offset = 0;
                }

                PullResult pullResult = consumer.pull(mq, "*", offset, 32);
                switch (pullResult.getPullStatus()) {
                    case FOUND:
                        System.out.printf("Found messages: %d%n", pullResult.getMsgFoundList().size());
                        if (pullResult.getMsgFoundList().size() > 0) {
                            System.out.println(pullResult.getMsgFoundList().get(0).getMsgId());
                            System.out.println(pullResult.getMsgFoundList().get(1).getMsgId());
                            System.out.println(pullResult.getMsgFoundList().get(2).getMsgId());
                            System.out.println("发送预警邮件");
                        }
                        offset = pullResult.getNextBeginOffset();
                        break;
                    case NO_NEW_MSG:
                        System.out.println("No new messages");
                        break;
                    case NO_MATCHED_MSG:
                        System.out.println("No matched messages");
                        break;
                    case OFFSET_ILLEGAL:
                        System.out.println("Offset illegal");
                        break;
                    default:
                        break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
